package com.csw.flink.source

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}

object Demo03MysqlSource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    //使用Mysqlsource
    val scoreDS: DataStream[Score] = env.addSource(new MysqlSource)

    val idNum: DataStream[(Int, Int)] = scoreDS.map(sco => (sco.studentId, sco.score)).keyBy(_._1).sum(1)

    idNum.print()

    env.execute()

  }

  case class Score(id: Int, subjectName: String, score: Int, studentId: Int)

  /**
    * 自定义source的接口
    *
    * SourceFunction  单线程source
    * RichSourceFunction  单线程source，  多了一个open和close方法用于打开和关闭连接
    * ParallelSourceFunction  并行的source  可以提高读取数据的效率
    * RichParallelSourceFunction  并行的source  多了一个open和close方法用于打开和关闭连接
    *
    */

  /**
    * 自定义source 读取mysql  是一个有界流
    */

  class MysqlSource extends RichSourceFunction[Score] {

    var conn: Connection = _

    override def open(parameters: Configuration): Unit = {
      /**
        * 读取mysql中的数据
        */
      Class.forName("com.mysql.jdbc.Driver")
      conn = DriverManager.getConnection("jdbc:mysql://master:3306/csw", "root", "123456")
    }

    override def run(sourceContext: SourceFunction.SourceContext[Score]): Unit = {

      val stat: PreparedStatement = conn.prepareStatement("select * from score")

      val resultSet: ResultSet = stat.executeQuery()

      while (resultSet.next()) {

        val id: Int = resultSet.getInt("id")
        val subjectName: String = resultSet.getString("subjectName")
        val score: Int = resultSet.getInt("score")
        val studentId: Int = resultSet.getInt("studentId")


        val score1: Score = Score(id, subjectName, score, studentId)

        /**
          * 将数据发送到下游
          *
          */
        sourceContext.collect(score1)

      }
    }

    /**
      * close 在run 之后执行 一般用户关闭链接
      *
      */
    override def close(): Unit = {
      conn.close()
    }

    override def cancel(): Unit = {

    }
  }

}
